package org.onion.pattern.event.springboot.rocketmq.publish;


import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.MQClientAPIImpl;
import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.header.namesrv.GetRouteInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQ;

import java.util.List;
import java.util.Map;

@Slf4j
@NoArgsConstructor
public class _Producer extends DefaultMQProducer {
    @Getter
    @Setter
    private boolean enabledAutoCreateTopic = true;

    public _Producer(RPCHook rpcHook) {
        super(rpcHook);
    }

    public _Producer(String producerGroup) {
        super(producerGroup, true, null);
    }

    public _Producer(String producerGroup, RPCHook rpcHook) {
        super(producerGroup, rpcHook, true, null);
    }

//    public _Producer(String producerGroup, RPCHook rpcHook,
//                     List<String> topics) {
//        super(producerGroup, rpcHook, topics);
//    }

    public _Producer(String producerGroup, boolean enableMsgTrace,
                     String customizedTraceTopic) {
        super(producerGroup, enableMsgTrace, customizedTraceTopic);
    }

    public _Producer(String producerGroup, RPCHook rpcHook,
                     boolean enableMsgTrace, String customizedTraceTopic) {
        super(producerGroup, rpcHook, enableMsgTrace, customizedTraceTopic);
    }

//    public _Producer(String producerGroup, RPCHook rpcHook,
//                     List<String> topics, boolean enableMsgTrace, String customizedTraceTopic) {
//        super(producerGroup, rpcHook, topics, enableMsgTrace, customizedTraceTopic);
//    }

    public _Producer(String namespace, String producerGroup) {
        super(namespace, producerGroup);
    }

    @Override
    public SendResult send(Message msg, MessageQueueSelector selector, Object arg, long timeout)
            throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        final String topic = withNamespace(msg.getTopic());
        autoCreateTopic(topic);
        msg.setTopic(topic);
        return this.defaultMQProducerImpl.send(msg, selector, arg, timeout);
    }


    protected void autoCreateTopic(String topicName) {
        if (!enabledAutoCreateTopic) {
            return;
        }
        final TopicPublishInfo topicPublishInfo = this.defaultMQProducerImpl.getTopicPublishInfoTable()
                .get(topicName);
        if (null == topicPublishInfo || !topicPublishInfo.ok()) {
            final MQClientAPIImpl mqClientAPIImpl = this.defaultMQProducerImpl.getmQClientFactory()
                    .getMQClientAPIImpl();

            final RemotingClient remotingClient = mqClientAPIImpl.getRemotingClient();
            GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();
            requestHeader.setTopic(topicName);
            RemotingCommand request = RemotingCommand
                    .createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, requestHeader);
            try {
                RemotingCommand response = remotingClient.invokeSync(null, request, 3 * 1000);
                if (response.getCode() == ResponseCode.TOPIC_NOT_EXIST) {
                    RemotingCommand cluster_request = RemotingCommand
                            .createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, null);
                    RemotingCommand cluster_response = remotingClient
                            .invokeSync(null, cluster_request, 3 * 1000);
                    if (cluster_response.getCode() == ResponseCode.SUCCESS) {
                        ClusterInfo clusterInfo = ClusterInfo
                                .decode(cluster_response.getBody(), ClusterInfo.class);

                        final Map<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();
                        for (BrokerData brokerData : brokerAddrTable.values()) {
                            TopicConfig topicConfig = new TopicConfig();
                            topicConfig.setWriteQueueNums(this.getDefaultTopicQueueNums());
                            topicConfig.setReadQueueNums(this.getDefaultTopicQueueNums());
                            topicConfig.setTopicName(topicName);
                            mqClientAPIImpl
                                    .createTopic(brokerData.selectBrokerAddr(),
                                            RocketMQ.AUTO_CREATE_TOPIC_KEY_TOPIC_BY_VERSION4_3, topicConfig,
                                            3 * 1000);
                            log
                                    .info("【事件】自动创建主题成功,cluster:{},broker:{},topic:{}"
                                            , brokerData.getCluster()
                                            , brokerData.getBrokerName()
                                            , topicName);
                        }
                    }
                }

            } catch (Exception e) {
                log.error("【事件】自动创建主题失败", e);
            }
        }

    }

}
